
#include <iostream>

#include <boost/thread/thread.hpp>
#include <boost/bind.hpp>

#include <boost/filesystem.hpp>
#include <boost/algorithm/string.hpp>

#include "boque/trasio/channel.hpp"
#include "boque/trasio/client.hpp"

namespace boque		{
namespace trasio	{


///////////////////////////////////////////////////////////////////////
//

class	Watcher
{
	typedef	void*	channel_v_ptr;
	typedef	std::map< std::string, channel_v_ptr >	channels_type;
	typedef	channels_type::iterator		channels_iterator_type;

public	:
	Watcher( void* channels_vp ) 
		:	_will_stop( false ),
			_running( true )
	{
		_channels_p = (channels_type *)channels_vp;
	}

	~Watcher()
	{
	}

public	:
	void	run( void )
	{
		_running = true;

		while ( !_will_stop ) {
			for ( channels_iterator_type it = _channels_p->begin();
				  _channels_p->end() != it;
				  ++it ) {
				Channel*	channel_p = (Channel *)it->second;

				if ( channel_p->inBufferedMode() ) {
					if ( !channel_p->isRelaying() ) {
						if ( NULL == channel_p->getWaiter() ) {
							boost::thread*	tp = new boost::thread( boost::bind(&Channel::relay, channel_p) );

							channel_p->setWaiter( tp );
						}

						channel_p->enableRelay();
					}
				}
				else if ( channel_p->inAliveMode() ) {
					if ( NULL != channel_p->getWaiter() ) {
						boost::thread*	tp = channel_p->getWaiter();

						tp->join();
						delete tp;

						channel_p->setWaiter( NULL );
					}
				}
			}

			boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
		}

		// stop all relaying
		for ( channels_iterator_type it = _channels_p->begin();
			  _channels_p->end() != it;
			  ++it ) {
			Channel*	channel_p = (Channel *)it->second;

			while ( channel_p->isRelaying() ) {
				channel_p->disableRelay();
				boost::this_thread::sleep(boost::posix_time::milliseconds(100));
			}
		}

		_running = false;
	}

	void	stop( void )
	{
		_will_stop = true;
	}

	bool	running( void )
	{
		return	_running;
	}
	
private	:
	channels_type*	_channels_p;

	bool	_will_stop;
	bool	_running;
};


///////////////////////////////////////////////////////////////////////
//

Client::Client( const std::string& buf_dir )
	:	_buf_dir( buf_dir ),
		_watcher_vp( NULL ),
		_watcher_tp( NULL )
{
	Watcher*	watcher_p = new Watcher( &_channels );
	_watcher_vp = watcher_p;

	_watcher_tp = new boost::thread( boost::bind(&Watcher::run, watcher_p) );
}

Client::~Client()
{
	if ( NULL != _watcher_vp ) {
		Watcher*	watcher_p = (Watcher *)_watcher_vp;

		while ( watcher_p->running() ) {
			boost::this_thread::sleep(boost::posix_time::milliseconds(100));
			watcher_p->stop();
		}

		delete watcher_p;

		_watcher_vp = NULL;
	}

	if ( NULL != _watcher_tp ) {
		_watcher_tp->join();
		delete _watcher_tp;

		_watcher_tp = NULL;
	}
}


///////////////////////////////////////////////////////////////////////
//

int
Client::send( const std::string& to, const std::string& data )
{
	channel_v_ptr	channel_vp = getChannel( to );
	Channel*	channel_p = static_cast< Channel* >( channel_vp );

	int	rc = channel_p->send( data );

	return	rc;
}

///////////////////////////////////////////////////////////////////////
//


Client::channel_v_ptr
Client::getChannel( const std::string& addr )
{
	channels_iterator_type	result = _channels.find( addr );
	if ( _channels.end() != result ) {
		// found
		return	result->second;
	}

	std::string	dir_s = addr;
	boost::replace_all( dir_s, ":", "%" );
	boost::replace_all( dir_s, "/", "%" );

	std::string	channel_dir = _buf_dir + "/" + dir_s;

	boost::filesystem::path	p( channel_dir );
	if ( !boost::filesystem::exists( p ) ) {
        try {
            boost::filesystem::create_directories( p );
        }   
        catch ( std::exception& e ) { 
            std::ostringstream  msg;
            msg << "fail to create channel buf directory " << channel_dir << " -- " << e.what();
            throw   std::runtime_error( msg.str() );
        }   
	}

	Channel*	channel_p = new Channel( addr, channel_dir );

	_channels.insert( std::make_pair( addr, channel_p ) );

	return	channel_p;
}


///////////////////////////////////////////////////////////////////////
//

}	//// namespace trasio
}	//// namespace boque

